package com.atguigu.flink.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * Created by 黄凯 on 2023/6/25 0025 16:25
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 */
public class Flink02_savepoint {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //备份的配置信息
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        //开启了checkpoint(备份)，每间隔2s自动备份。程序就会无限次重启
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        env
                .socketTextStream("127.0.0.1", 8888)
                .map(new MyMapFunction())
                .addSink(new SinkFunction<String>() {
                    @Override
                    public void invoke(String value, Context context) throws Exception {
                        if (value.contains("x")) {
                            //人为手动抛出一个异常，为了模拟故障的情况
                            throw new RuntimeException("出异常了!");
                        }
                        System.out.println(value);
                    }
                });


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /*
        功能： 实现字符串的累积打印

            Flink提供的状态就是拥有自动备份，自动恢复功能的集合。
     */
    public static class MyMapFunction implements MapFunction<String, String>, CheckpointedFunction {

        //当集合用，本质是ManagedState中的OpearateState
        ListState<String> strs;

        @Override
        public String map(String s) throws Exception {
            //存
            strs.add(s);
            //取
            return strs.get().toString();
        }

        /*
            周期性(默认200ms，可以调整)将状态以快照的形式进行备份
         */
        @Override
        public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
            System.out.println("MyMapFunction.snapshotState");
            //备份是自动进行的，无需进行任何手动操作
        }

        /*
            初始化： 在第一次启动或任务失败重启后执行，从之前的备份中找到状态，重新赋值。
            FunctionInitializationContext context: 程序的运行环境，或上下文，可以从中获取很多额外的信息。
         */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyMapFunction.initializeState");
            //找到备份状态的存储设备
            OperatorStateStore operatorStateStore = context.getOperatorStateStore();
            //从备份中找到之前备份的变量
            ListStateDescriptor<String> listStateDescriptor = new ListStateDescriptor<>("strs", String.class);
            strs = operatorStateStore.getListState(listStateDescriptor);
        }
    }

}
